package com.hujf.flink.DataStreamAPI;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple6;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.JoinedStreams;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

import java.util.concurrent.TimeUnit;

/**
 * @Author hujf
 * @Date 2022/3/24 14:44
 * @Description TODO
 */
public class Union {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment =  StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Tuple3<Integer, String, Double>> tuple3DataStreamSource = environment.fromElements(new Tuple3<>(1, "BMW", 1.0),
                new Tuple3<>(2, "BMW", 2.0),
                new Tuple3<>(3, "TSL", 3.0),
                new Tuple3<>(4, "TSL", 4.0));


        DataStreamSource<Tuple3<Integer, String, Double>> tuple3DataStreamSource2 = environment.fromElements(new Tuple3<>(1, "BMW1", 1.0),
                new Tuple3<>(2, "BMW", 2.0),
                new Tuple3<>(3, "TSL", 34.0),
                new Tuple3<>(4, "TSL", 4.0));

        DataStream<Object> apply = tuple3DataStreamSource.join(tuple3DataStreamSource2).where(o -> o.f0).equalTo(o -> o.f0).window(
                TumblingProcessingTimeWindows.of(Time.seconds(1)
                )).apply(new JoinFunction<Tuple3<Integer, String, Double>, Tuple3<Integer, String, Double>, Object>() {
            @Override
            public Object join(Tuple3<Integer, String, Double> first, Tuple3<Integer, String, Double> second) throws Exception {
                if (first.f1 != second.f1) {
                    return new Tuple6<>(first.f0, second.f0, first.f1, second.f1, first.f2, second.f2);
                }

                if (!first.f2.equals(second.f2)) {
                    return new Tuple6<>(first.f0, second.f0, first.f1, second.f1, first.f2, second.f2);
                }
                return null;
            }
        });
        apply.print();

        environment.execute("123");
    }
}
